/**
 * @file Thread.cpp
 * @author zeppyrz (zeppyrz@gmail.com)
 * @brief Thread class implementation
 * @version 0.1
 * @date 2025-11-03
 *
 * @copyright Copyright (c) 2025
 *
 */

#include "Thread.h"

#include <event2/event.h>

#include <iostream>

#include "Task.h"
#ifndef _WIN32
#  include <unistd.h>
#endif

namespace ZeDisk {

static void _ClearTaskCallback(evutil_socket_t fd, short arg, void *p) {
  auto thread = static_cast<Thread *>(p);
  thread->ClearTaskCallback(fd, arg);
}

// 激活线程回调函数
static void notify_cb(evutil_socket_t fd, short what, void *arg) {
  auto *t = static_cast<Thread *>(arg);
  // 水平触发
  t->Notify(fd, what);
}

Thread::Thread(int id)
    : id_(id),
      running_(false),
      joined_(false),
      notify_send_fd(-1),
      base_(nullptr) {}

Thread::~Thread() {
  event_base_loopbreak(base_);
  if (running_ || !joined_) {
    thread_.detach();
  }
}

bool Thread::Start() {
  if (!SetUp()) {
    std::cout << "[ERROR] Thread SetUp() failed" << std::endl;
    return false;
  }
  // 启动线程
  running_ = true;
  thread_ = std::move(std::thread(&Thread::Main, this));
  return true;
}

void Thread::Stop() {
  running_ = false;
}

void Thread::Main() {
#ifdef DEBUG
  std::cout << "[DEBUG] Thread::Main() id: " << id_ << std::endl;
#endif
#if 0
  // 开启base事件分发
  event_base_dispatch(base_);
#else
  while (running_) {
    event_base_loop(base_, EVLOOP_NONBLOCK);
    std::this_thread::sleep_for(std::chrono::milliseconds(1));
  }
#endif
  event_base_free(base_);
#ifdef DEBUG
  std::cout << "[DEBUG] Leave Thread::Main()" << std::endl;
#endif
}

bool Thread::SetUp() {
  // windows 使用配对socket，linux使用管道
#ifdef _WIN32
  // 创建一个socket pair 可以互相通信 fds[0] read, fds[1] write
  evutil_socket_t fds[2];
  if (0 > evutil_socketpair(AF_INET, SOCK_STREAM, 0, fds)) {
    std::cout << "[ERROR] evutil_socketpair() failed" << std::endl;
    return false;
  }
  evutil_make_socket_nonblocking(fds[0]);
  evutil_make_socket_nonblocking(fds[1]);
#else
  // 创建的管道 使用read write 读取
  int fds[2];
  if (pipe(fds)) {
    std::cout << "[ERROR] evutil_pipe() failed" << std::endl;
    return false;
  }
#endif
  // 读取绑定到event事件中，写入要保存
  notify_send_fd = fds[1];

  // 创建libevent上下文
  event_config *cfg = event_config_new();
  event_config_set_flag(cfg, EVENT_BASE_FLAG_NOLOCK);  // 无锁
  base_ = event_base_new_with_config(cfg);
  event_config_free(cfg);
  if (!base_) {
    std::cout << "[ERROR] event_base_new_with_config() failed" << std::endl;
    return false;
  }

  // 添加管道监听事件
  event *ev = event_new(base_, fds[0], EV_READ | EV_PERSIST, notify_cb, this);
  event_add(ev, nullptr);

  // 添加清理计时器
  clear_task_timer_ =
      event_new(base_, -1, EV_PERSIST, _ClearTaskCallback, this);
  timeval tv = {0, 0};
  tv.tv_sec = 15;
  event_add(clear_task_timer_, &tv);

  return true;
}

void Thread::Join() {
  joined_ = true;
  thread_.join();
}

void Thread::Notify(evutil_socket_t fd, short what) {
  char buf[2];
#ifdef _WIN32
  auto re = ::recv(fd, buf, 1, 0);
#else
  auto re = ::read(fd, buf, 1);
#endif
  if (re < 0) {
    std::cout << "[ERROR] read() failed" << std::endl;
    return;
  }

#ifdef DEBUG
  std::cout << "[DEBUG] notify thread: " << id_ << std::endl;
#endif

  // 获取任务 并初始化任务
  {
    std::lock_guard<std::mutex> lock(tasks_mtx_);
    if (tasks_.empty()) return;

    auto task = tasks_.front();
    tasks_.pop_front();
    task->base() = base_;
    task->Init();
    activated_tasks_.push_back(task);
  }
}

void Thread::Activate() {
#ifdef _WIN32
  auto re = ::send(notify_send_fd, "c", 1, 0);
#else
  auto re = ::write(notify_send_fd, "c", 1);
#endif
  if (re < 0) {
    std::cout << "[ERROR] write() failed" << std::endl;
    return;
  }
#ifdef DEBUG
  std::cout << "[DEBUG] activate thread: " << id_ << std::endl;
#endif
}

void Thread::AddTask(std::shared_ptr<Task> task) {
  if (!task) return;
  task->base() = this->base_;
  std::lock_guard<std::mutex> lock(tasks_mtx_);
  tasks_.push_back(task);
}

void Thread::ClearTaskCallback(evutil_socket_t, short) {
  auto presize = activated_tasks_.size();
  activated_tasks_.remove_if([](auto &&it) { return it->Completed(); });
#ifdef DEBUG
  std::cout << "[DEBUG] Thread " << id_
            << " ClearTask count: " << presize - activated_tasks_.size()
            << std::endl;
#endif
}

}  // namespace ZeDisk